package com.elinshaw.pattern.bus;

import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;

public class Dispatcher {

    private final Executor executor;

    private final EventExceptionHandle eventExceptionHandle;

    public final static Executor SEQ_EXECUTOR_SERVICE = SeqExecutorService.INSTANCE;

    public final static Executor PER_THREAD_EXECUTOR_SERVICE = PerThreadExecutorService.INSTANCE;

    public Dispatcher(Executor executor, EventExceptionHandle eventExceptionHandle) {
        this.executor = executor;
        this.eventExceptionHandle = eventExceptionHandle;
    }

    public void dispatcher(Bus bus, Registry registry, Object event, String topic) {
        ConcurrentLinkedQueue<Subscriber> subscribers = registry.scanSubscriber(topic);
        if (subscribers == null) {
            if (eventExceptionHandle != null) {
                eventExceptionHandle.handle(new IllegalArgumentException(bus.getBusName() + "not bind"), new BasicEventContext(bus.getBusName(), null, event));
            }
            return;
        }
        subscribers.stream().filter(subscriber -> !subscriber.isDisable()).filter(
                subscriber -> subscriber.getSubscriberMethod().getParameterTypes()[0].isAssignableFrom(event.getClass())
        ).forEach(subscriber -> realInvokeSubscribe(subscriber, event, bus));
    }

    private void realInvokeSubscribe(Subscriber subscriber, Object event, Bus bus) {
        executor.execute(() -> {
            try {
                subscriber.getSubscriberMethod().invoke(subscriber.getSubscriberObject(), event);
            } catch (Exception e) {
                if (eventExceptionHandle != null) {
                    eventExceptionHandle.handle(e, new BasicEventContext(bus.getBusName(), subscriber, event));
                }
            }
        });
    }

    public static Dispatcher newDispatcher(EventExceptionHandle eventExceptionHandle, Executor executor) {
        return new Dispatcher(executor, eventExceptionHandle);
    }


    public static Dispatcher seqDispatcher(EventExceptionHandle eventExceptionHandle) {
        return new Dispatcher(SEQ_EXECUTOR_SERVICE, eventExceptionHandle);
    }


    public static Dispatcher perThreadDispatcher(EventExceptionHandle eventExceptionHandle) {
        return new Dispatcher(PER_THREAD_EXECUTOR_SERVICE, eventExceptionHandle);
    }

    public void close() {
        if (executor instanceof ExecutorService) {
            ((ExecutorService) executor).shutdown();
        }
    }


}
